import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.apache.hadoop.io.BooleanWritable;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;
import java.util.concurrent.Future;
public class Hive2Kafka extends GenericUDF {
    @Override
    public ObjectInspector initialize(ObjectInspector[] objectInspectors) throws UDFArgumentException {
        if (objectInspectors.length != 3){
            throw  new UDFArgumentException("参数个数不正确！");
        }
        return PrimitiveObjectInspectorFactory.writableBooleanObjectInspector;
    }

    @Override
    public Object evaluate(DeferredObject[] deferredObjects) throws HiveException {
        String topic = deferredObjects[0].get().toString();
        String rowKey=deferredObjects[1].get().toString();
        String value = deferredObjects[2].get().toString();
        Properties properties = new Properties();
        try {
            InputStream inputStream = new FileInputStream("/opt/Producer.properties");
            properties.load(inputStream);
        } catch (IOException e) {
            System.out.println("读取配置文件有误！");
            e.printStackTrace();
        }
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, rowKey, value);
        Future<RecordMetadata> future = kafkaProducer.send(producerRecord);
        kafkaProducer.close();
        BooleanWritable booleanWritable = new BooleanWritable(future.isDone());
        return booleanWritable;
    }

    @Override
    public String getDisplayString(String[] strings) {
        return "Hive2Kafka";
    }
}
